package com.peng.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;

import static java.lang.Thread.sleep;

//高级消费者
public class CustomerConsumer {
    public static void main(String[] args) {
        //1 配置信息
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");//Kafka集群
        props.put("group.id", "test");//消费者组id
        props.put("enable.auto.commit", "true");//自动提交 ，提交offset
        props.put("auto.commit.interval.ms", "1000");//提交延时1s

        //如果要从头消费数据加上下面两条
        /*props.put("group.id","hdfjd");//设置消费者组id，每次要该改名
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");//将offset偏移量设成最早
        */

        //kv反序列化
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        //2 创建消费者
        KafkaConsumer<String,String> consumer=new KafkaConsumer<String, String>(props);

        //3 指定topic
        consumer.subscribe(Arrays.asList("first","first-topic","second"));
        //下面两句可以实现指定分区 指定offset 指定topic
        //consumer.assign(Collections.singletonList(new TopicPartition("first",0)));
        //consumer.seek(new TopicPartition("first",0),2);

        //4 获取数据
        while(true){
            ConsumerRecords<String,String> consumerRecords=consumer.poll(100);
            for(ConsumerRecord<String,String> record:consumerRecords){
                System.out.println(record.topic()+"--"+record.partition()+"--"+record.value());

            }
        }
    }
}
